package com.fire.mongo.lessons;

import java.util.ArrayList;
import java.util.List;

import com.mongodb.client.AggregateIterable;
import com.mongodb.client.model.Accumulators;
import com.mongodb.client.model.Aggregates;
import com.mongodb.client.model.BsonField;
import com.mongodb.client.model.Facet;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.Sorts;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.junit.Assert;
import org.junit.Test;

import org.springframework.boot.test.context.SpringBootTest;

/**
 * @see Facet
 * @see Accumulators
 * @see Aggregates
 * 著作权归https://pdai.tech所有。
 * 链接：https://pdai.tech/md/db/nosql-mongo/mongo-x-usage-2.html
 *
 * $project：修改输入文档的结构。可以用来重命名、增加或删除域，也可以用于创建计算结果以及嵌套文档。
 * $match：用于过滤数据，只输出符合条件的文档。
 * $match使用MongoDB的标准查询操作。
 * $limit：用来限制MongoDB聚合管道返回的文档数。
 * $skip：在聚合管道中跳过指定数量的文档，并返回余下的文档。
 * $unwind：将文档中的某一个数组类型字段拆分成多条，每条包含数组中的一个值。
 * $group：将集合中的文档分组，可用于统计结果。
 * $sort：将输入文档排序后输出。
 * $geoNear：输出接近某一地理位置的有序文档。
 * $bucket: 分组（分桶）计算。
 * $facet : 多次分组计算。
 * $out: 将结果集输出，必须是Pipline最后一个Stage。
 */
@SpringBootTest
public class UsingAggregationBuilders extends AbstractLesson {

  /*
  本节中将使用 Java Driver aggregation builders 去构建复杂的聚合 framework pipelines
   */

  @Test
  public void singleStageAggregation() {
    /*
    首先，我们将会创建单个聚合步骤，在该案例中，我们将会知道有多少葡萄牙的电影被拍摄
     */
    String country = "Portugal";

    /*
    我们将会使用MongoDB 的聚合框架。为此，我需要定义一个pipeline，它将持有所有的聚合步骤。
    在该案例中，我们的 pipeline 中仅有一个 $match 步骤，用mongo shell 可以表示如下
    db.movies.aggregate([{$match: {countries: "Portugal"}}])
     */

    // 表达匹配标准
    Bson countryPT = Filters.eq("countries", country);

    /*
    aggregation（）collection方法获取定义管道不同阶段的Bson对象列表，
    我们将使用ArrayList设置这些对象。
     */

    List<Bson> pipeline = new ArrayList<>();

    /*
    您应该使用Aggregates builder类，而不是手动构造表示聚合阶段的Bson文档。
    com.mongodb.client.model.Aggregates 为每个支持聚合的阶段提供了一组语法糖类生成器和方法。
    尽管我们可以通过在文档或Bson对象中添加相应的表达式来构建任何聚合阶段，
    但Aggregates 允许更简洁的阶段构建，只需较少的键入。
    match（）方法将过滤器表达式作为参数，类似于我们在find（）命令中使用的过滤器表达式。
    */

    Bson matchStage = Aggregates.match(countryPT);

    //添加 matchStage 到 pipeline 中
    pipeline.add(matchStage);

    /*
    一旦我们将 match 步骤添加到pipeline 数组中，我们就可以执行aggregate() 集合的命令
    */

    AggregateIterable<Document> iterable = moviesCollection.aggregate(pipeline);

    /*
     aggregate() 方法的结果中，我们可以得到 AggregateIterable。和其他的迭代相似。
     该对象允许我们迭代结果集
    */

    //将所有电影收集到数据列表中
    List<Document> builderMatchStageResults = new ArrayList<>();
    iterable.into(builderMatchStageResults);

    /*
    上述步骤将会列出葡萄牙生产了115条电影
     */
    Assert.assertEquals(115, builderMatchStageResults.size());
  }

  @Test
  public void aggregateSeveralStages() {
    /*
    使用find（）命令可以实现单个聚合管道，尤其是$match阶段。
    让我们使用更有趣的东西，这正是我们应该使用聚合框架的目的。
     */

    List<Bson> pipeline = new ArrayList<>();

    /*
    对于所有在葡萄牙制作的电影，获取某个演员演出次数的总和。
    一个演员有多少次参与这个国家制作的电影。当然，我们也不要忘记返回按演出数量升序排序的结果。
    在mongo shell中，这个问题可以通过以下汇总来回答：
    db.movies.aggregate([
        {$match: {countries: "Portugal"}},
        {$unwind: "$cast"},
        {$group: {_id: "$cast", gigs: {$sum: 1}}}
    ])
    */

    /*
    - $match to find all movies produced in portugal
     */
    String country = "Portugal";
    Bson countryPT = Filters.eq("countries", country);
    Bson matchStage = Aggregates.match(countryPT);

    /*
    使用 $unwind 将 $cast 数据进行解绑
     */
    Bson unwindCastStage = Aggregates.unwind("$cast");

    /*
    使用$group 按照演员名字进行分组，并且统计结果集中演员数量出现的次数
     */
    String groupIdCast = "$cast";

    /*
    分组操作可以对聚集操作做一些排序，这些操作有 $sum, $avg, $min, $max，这些都是非常
    好的分组操作。
     */

    //使用$sum累加器为每个演员出现次数加1。
    BsonField sum1 = Accumulators.sum("count", 1);

    // 添加组id和累加器
    Bson groupStage = Aggregates.group(groupIdCast, sum1);

    /*
    - 对出场次数进行排序
     */
    Bson sortOrder = Sorts.descending("count");

    Bson sortStage = Aggregates.sort(sortOrder);

    /*
    有了所有这些阶段，我们现在可以用比单个$match阶段更复杂的管道调用聚合方法了。
     */

    pipeline.add(matchStage);
    pipeline.add(unwindCastStage);
    pipeline.add(groupStage);
    pipeline.add(sortStage);

    AggregateIterable<Document> iterable = moviesCollection.aggregate(pipeline);

    List<Document> groupByResults = new ArrayList<>();
    for (Document doc : iterable) {
      System.out.println(doc);
      groupByResults.add(doc);
    }

    /*
    聚合框架还提供了将通常由几个阶段表示的操作组合在一起的阶段。
    例如，$sortByCount将$group和$sum累加器与$sort阶段结合起来。
    不相信我？好吧，我们来看看！
     */

    List<Bson> shorterPipeline = new ArrayList<>();

    shorterPipeline.add(matchStage);
    shorterPipeline.add(unwindCastStage);

    // 调用 $sortByCountStage
    Bson sortByCount = Aggregates.sortByCount("$cast");

    // 将 $sortByCount 步骤添加到 shortPipeline
    shorterPipeline.add(sortByCount);

    List<Document> sortByCountResults = new ArrayList<>();

    for (Document doc : moviesCollection.aggregate(shorterPipeline)) {
      System.out.println(doc);
      sortByCountResults.add(doc);
    }

    /*
    Running both pipelines, the same set of results.
     */

    Assert.assertEquals(groupByResults, sortByCountResults);
  }

  @Test
  public void complexStages() {

    /*
     并非所有聚合阶段都是相等的。就操作类型和操作参数而言，有些操作比其他操作更复杂。
     例如：$lookup阶段比$addFields阶段需要执行更多的参数/选项
     {
        $lookup: {
            from: "collection_name",
            pipeline: [{}] - sub-pipeline
            let: {...} - expression
            as: "field_name" - output array field name
        }
     }

     vs

     {
        $addFields: {
            "new_field": {expression} - expression that computes field value
           }
     }

     */

    List<Bson> pipeline = new ArrayList<>();

    /*
    为了举例说明这个场景，让我们继续做以下操作：

    -电影文档的创建方面，找出西班牙的电影

    -演员方面：在葡萄牙制作的电影中找到的演员名单

    -体裁方面的计数：体裁列表及其计数

    -年桶上的刻面：匹配的电影年

    对于每个方面，我们将创建一个com.mongodb.client.Facet 对象
     */

    // $unwind the cast array
    Bson unwindCast = Aggregates.unwind("$cast");

    // create a set of cast members with $group
    Bson groupCastSet = Aggregates.group("", Accumulators.addToSet("cast_list", "$cast"));

    /*
    Facet constructor takes a facet name and variable arguments,
    variable-length argument, of sub-pipeline stages that build up the
    expected facet values.
    For the cast_filter we need to unwind the cast arrays and use group
    to create a set of cast members.
     */

    Facet castMembersFacet = new Facet("cast_members", unwindCast, groupCastSet);

    // unwind genres
    Bson unwindGenres = Aggregates.unwind("$genres");

    // genres facet bucket
    Bson genresSortByCount = Aggregates.sortByCount("$genres");

    // create a genres count facet
    Facet genresCountFacet = new Facet("genres_count", unwindGenres, genresSortByCount);

    // year bucketAuto
    Bson yearBucketStage = Aggregates.bucketAuto("$year", 10);

    // year bucket facet
    Facet yearBucketFacet = new Facet("year_bucket", yearBucketStage);

    /*
    The Aggregates.facet() method also takes variable set of Facet
    objects that composes the sub-pipelines of each facet element.

    db.movies.aggregate([
        { "$match" : { "countries" : "Portugal" } },
        { "$facet" : {
            "cast_members" : [{ "$unwind" : "$cast" }, { "$group" : { "_id" : "", "cast_list" : { "$addToSet" : "$cast" } } }],
            "genres_count" : [{ "$unwind" : "$genres" }, { "$sortByCount" : "$genres" }],
            "year_bucket" : [{ "$bucketAuto" : { "groupBy" : "$year", "buckets" : 10 } }]
            }
        }
      ])
     */

    // $facets stage
    Bson facetsStage = Aggregates.facet(castMembersFacet, genresCountFacet, yearBucketFacet);

    // match stage
    Bson matchStage = Aggregates.match(Filters.eq("countries", "Portugal"));

    // putting it all together
    pipeline.add(matchStage);
    pipeline.add(facetsStage);

    int countDocs = 0;
    for (Document doc : moviesCollection.aggregate(pipeline)) {
      System.out.println(doc);
      countDocs++;
    }

    Assert.assertEquals(1, countDocs);
  }

  /*
  Let's recap:
  - Aggregation framework pipelines are composed of lists of Bson stage
  document objects
  - Use the driver Aggregates builder class to compose the different stages
  - Use Accumulators, Sorts and Filters builders to compose the different
  stages expressions
  - Complex aggregation stages can imply several different sub-pipelines
  and stage arguments.
   */

}
